/*
 Copyright (C) 2011  Matteo Magnani, Gabriele Nunziante
 
 This program is free software: you can redistribute it and/or modify
 it under the terms of the GNU General Public License as published by
 the Free Software Foundation, either version 3 of the License, or
 (at your option) any later version.
 
 This program is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 GNU General Public License for more details.
 
 You should have received a copy of the GNU General Public License
 along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

/******************************************************************************************************
 * CRS : Server to store tweets into the database and index them                                      *
 ******************************************************************************************************
 * 1) Reads from clients one of the following commands:
 *
 * tweet\n
 * tw_id \t tw_id_user \t tw_text \t tw_time \t tw_reply_to \t tw_origin \t tw_n_retweet \t tw_lang \n
 *
 * OR
 * 
 * user\n
 * usr_id \t usr_name \t usr_screen_name \t usr_image \t usr_followers \n
 *
 *
 * The first time it must be executed with the -c flag, determining the creation of the index.
 ******************************************************************************************************/

import java.io.*;
import java.net.*;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.sql.SQLException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.StringTokenizer;
import java.util.Vector;
import java.util.ArrayList;
import java.util.Properties;

import org.apache.lucene.analysis.*;
import org.apache.lucene.document.*;
import org.apache.lucene.index.*;
import org.apache.lucene.store.*;
import org.apache.lucene.search.*;
import org.apache.lucene.queryParser.*;
import org.apache.lucene.util.Version;

public class CRStoreServer extends Thread {
	
	/* DEBUG variables */
	boolean q = false; /* true to print queries */
	boolean t = false; /* true to print tweets */
	boolean s = false; /* true to print system messages */
	
	/* Configuration variables [read from file crservers.cfg] */
	private static int port; // Server listening on this port
	private static String indexDir; // directory where the index will be stored	
	private static String dbServerAddress; // Database address
	private static String dbName; // Database name
	private static String dbUser; // Database user
	private static String dbPassword; // Database password
	
	//private URLConnection conn;
	private static ServerSocket serversocket;
	
	/* DB connection */
	private static Database db;
	
	/**
	 * @param if executed for the first time, specify -c to create the index
	 */
	public static void main(String[] args) throws Exception
	{
		/* Checks if IR index should be created */
		boolean create = false;
		if (args.length>0)
			if (args[0].equals("-c"))
				create = true;
		
		/* Reads configuration */
		Properties p = new Properties();
		try {
			p.load(new FileReader("crservers.cfg"));
		}
		catch (IOException ex) {
			//System.err.println("[ERR] " + ex.getMessage());
			System.err.println("Cannot open config file crservers.cfg");
			System.exit(-1);
		}
		try {
			port = Integer.parseInt(p.getProperty("store-port"));
		}
		catch (Exception ex) {
			//System.err.println("[ERR] " + ex.getMessage());
			System.err.println("<store-port> config info not found");
			System.exit(-1);
		}
		try {
			indexDir = p.getProperty("indexDir");
			if (indexDir==null) throw new Exception("Could not read <indexDir> config parameter");
			dbServerAddress = p.getProperty("dbServerAddress");
			if (dbServerAddress==null) throw new Exception("Could not read <dbServerAddress> config parameter");
			dbName = p.getProperty("dbName");
			if (dbName==null) throw new Exception("Could not read <dbName> config parameter");
			dbUser = p.getProperty("dbUser");
			if (dbUser==null) throw new Exception("Could not read <dbUser> config parameter");
			dbPassword = p.getProperty("dbPassword");
			if (dbPassword==null) throw new Exception("Could not read <dbPassword> config parameter");
		}
		catch (Exception ex) {
			System.err.println("[ERR] " + ex.getMessage());
			System.exit(-1);
		}
		
		/* starts the IR index */
		try { 
			IndexWriter writer = new IndexWriter(FSDirectory.open(new File(indexDir)), 
									 new SimpleAnalyzer(Version.LUCENE_31),create, IndexWriter.MaxFieldLength.UNLIMITED);	
			writer.close();
		}
		catch (Exception ex) {
			System.err.println("[ERR] " + ex.getMessage());
			System.err.println("Cannot setup LUCENE index");
			System.exit(-1);
		}
		
		/* Getting database connection */
		db = new Database(dbServerAddress, dbName, dbUser, dbPassword);
		if(!db.open()) {
			System.out.println("[ERR] DB Connection error: " + db.getError());
			System.exit(-1);
		}
		
		/* Opens the server socket */
		try {
			serversocket = new ServerSocket(port);
		}
		catch (Exception ex) {
			System.err.println("[ERR] " + ex.getMessage());
			System.err.println("Cannot start the server on port " + port);
			System.exit(-1);
		}
		
		System.out.println("\n* CRStoreServer waiting on port " + port);
		
		new CRStoreServer();
	}
	
	public CRStoreServer() throws Exception {
		while (true) {
			// Wait for connection and process it
			try {
				Socket client = serversocket.accept();
				client.setSoTimeout(5000);
				if (s) System.out.println("[CRStoreServer] Processing connection from " + client.getInetAddress());
				IndexSearcher searcher = new IndexSearcher(FSDirectory.open(new File(indexDir)));
				IndexWriter writer = new IndexWriter(FSDirectory.open(new File(indexDir)), 
										 new SimpleAnalyzer(Version.LUCENE_31),false, IndexWriter.MaxFieldLength.UNLIMITED);		
				Connect c = new Connect(client, db, writer, searcher);
				c.processConnection();
				writer.close();
				searcher.close();
				client.close();
			} catch(Exception e) { 
				System.err.println("[ERR] " + e.getMessage());
				e.printStackTrace();
			}
		}        
	}
}

/* Class to manage a client connection */
class Connect
{	
	/* DEBUG variables */
	boolean q = false; /* true to print queries */
	boolean t = false; /* true to print tweets */
	boolean s = false; /* true to print system messages */	
	
	/*  */
	private int tw_numFields = 9;
	private int tw_id = 0;
	private int tw_user_name = 1;
	private int tw_id_user = 2;
	private int tw_text = 3;
	private int tw_time = 4;
	private int tw_reply_to = 5;
	private int tw_origin = 6;
	private int tw_n_retweet = 7;
	private int tw_lang = 8;
	
	private int usr_id = 0;
	private int usr_name = 1;
	private int usr_screen_name = 2;
	private int usr_image = 3;
	private int usr_followers = 4;
	//private int usr_last_update = 5;
	
	/*  */
	URLConnection conn;
	private Socket client = null;
	private BufferedReader in = null;
	private PrintStream out = null;
	private IndexWriter writer;
	private IndexSearcher searcher;
	private Database db;
	private SimpleDateFormat sdf;
	
	public Connect(Socket clientSocket, Database db, IndexWriter writer, IndexSearcher searcher) {		
		this.client = clientSocket;
		this.db = db;
		this.writer = writer;
		this.searcher = searcher;
		
		sdf = new SimpleDateFormat("yyyy-MM-dd kk:mm:ss");
		
		try {
			/* input/output */
			in = new BufferedReader(new InputStreamReader(client.getInputStream()));
			out = new PrintStream(client.getOutputStream(), true);
		}
		catch (Exception ex) {
			System.out.println(ex.getMessage());
			return;
		}
	}
	
	public void processConnection() {
		try {
			String cmd = null;
			boolean res = false;
			cmd = in.readLine();
			
			if (cmd.equals("tweet")) {
				cmd = in.readLine();
				try {res = insertTweet(cmd);} catch (Exception e) {
					if (t) System.out.println("[CRStoreServer] could not insert tweet: " + e.getMessage() + " " + cmd);
				}
				if (res) if (t) System.out.println("[CRStoreServer] tweet inserted [" + cmd + "]");
				/*if (res) 
					out.println("OK");
				else 
						out.println("KO");*/
			}
			else if (cmd.equals("user")) {
				cmd = in.readLine();
				try {res = insertUser(cmd);} catch (Exception e) {
					if (t) System.out.println("[CRStoreServer] could not insert user: " + e.getMessage() + " " + cmd);
				}
				if (res) if (t) System.out.println("[CRStoreServer] user inserted [" + cmd + "]");
					/*if (res) 
						out.println("OK");
					else 
						out.println("KO");
				}	*/
			}
			else {
				System.err.println("[ERR] Cannot process instruction " + cmd);
				//out.println("KO");
				//break;
			}
				//cmd = in.readLine();
			//}
			//if (cmd.equals("stop")) if (s) System.out.println("[CRStoreServer] end of connection OK");
		} catch (Exception e) {
			e.printStackTrace();
			if (s) System.out.println("[CRStoreServer] connection lost");
		}
		
		try {
			out.close();
			in.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * @return true if tweet has been inserted 
	 */
	public boolean insertTweet(String record) throws Exception {
		String query;
		//System.out.println("tweet: " + record);
		String[] r = record.split("\t");
		// language check
		if (!r[tw_lang].equals("en") && !r[tw_lang].equals("it") && !r[tw_lang].equals("es") && !r[tw_lang].equals("de") && !r[tw_lang].equals("fr") && !r[tw_lang].equals("pt"))
			return false;
		// check if tweet is present
		query  = "SELECT id, id_user FROM tweet WHERE id='"+r[tw_id]+"'";
		if (q) System.out.println("[Q] " + query);
		ResultSet rs = db.executeQuery(query);		
		if ( rs!=null && rs.next() ) {
			if (rs.getString("id_user").trim().equals("-1") && !r[tw_id_user].equals("-1")) {
				query = "UPDATE tweet SET id_user='" + r[tw_id_user] + "', origin='" + r[tw_origin] + "' WHERE id='"+r[tw_id]+"'";
				if (q) System.out.println("[Q] " + query);
				db.executeUpdate(query);
			}
			return false; // tweet already present and complete
		}
		
		
			// insert tweet
			query = "INSERT DELAYED INTO tweet (id, user_name, id_user, text, time, reply_to, origin, n_retweet)" +
			" VALUES ('"+r[tw_id]+"', '"+formatString(r[tw_user_name])+"', '"+r[tw_id_user]+"', '"+formatString(r[tw_text])+"', " +
			"'"+formatDate(r[tw_time])+"', '"+r[tw_reply_to]+"', '"+r[tw_origin]+"', '"+r[tw_n_retweet]+"')";
			if (q) System.out.println("[Q] " + query);
			db.executeUpdate(query);
			
			// get user popularity
			int pop = 0;
			query = "SELECT followers FROM user WHERE id ='" + r[tw_id_user] + "'";
			rs = db.executeQuery(query);		
			if( rs!=null && rs.next() ) {
				pop = rs.getInt(1);
			}
			
			
			// get conversation metadata
		String conversationID = r[tw_origin];
		if (r[tw_origin].equals("-1")) conversationID = r[tw_id];
			Term idTerm = new Term("id",conversationID+"");
			Query q = new TermQuery(idTerm);
			TopDocs docs = searcher.search(q, 1);
			
			Document doc = new Document();
			
			if (docs.scoreDocs.length == 0) {
				// insert new conversation
				doc.add(new Field("id", conversationID+"", Field.Store.YES, Field.Index.NOT_ANALYZED_NO_NORMS));
				doc.add(new Field("num_tweets", "1", Field.Store.YES, Field.Index.ANALYZED));
				doc.add(new Field("text", r[tw_text], Field.Store.YES, Field.Index.ANALYZED));
				doc.add(new Field("den", "0", Field.Store.YES, Field.Index.NO));
				doc.add(new Field("pop", pop+"", Field.Store.YES, Field.Index.NO));
				doc.add(new Field("n_retweets", r[tw_n_retweet]+"", Field.Store.YES, Field.Index.NO));
				doc.add(new Field("start", formatDate(r[tw_time]), Field.Store.YES, Field.Index.NO));
				doc.add(new Field("end", formatDate(r[tw_time]), Field.Store.YES, Field.Index.NO));
				//System.out.print("+ " + writer.maxDoc() + " + " + r[tw_text]);
				writer.addDocument(doc);
				//System.out.println(" --> " + writer.maxDoc());
			}
			else {
				// get metadata
				Document oldDoc = searcher.doc(docs.scoreDocs[0].doc);
				int numTweets = Integer.parseInt(oldDoc.get("num_tweets"));
				int n_retweets = Integer.parseInt(oldDoc.get("n_retweets"));
				int oldPop = Integer.parseInt(oldDoc.get("pop"));					
				String start = oldDoc.get("start");								
				String end = oldDoc.get("end");			
				String text = oldDoc.get("text");
				String tweetTime = formatDate(r[tw_time]);
				
				if (differenceDate(tweetTime,end)<0) end = tweetTime;
				else if (differenceDate(start,tweetTime)<0) start = tweetTime;
				
				doc.add(new Field("id", conversationID+"", Field.Store.YES, Field.Index.NOT_ANALYZED_NO_NORMS));
				doc.add(new Field("num_tweets", (numTweets+1)+"", Field.Store.YES, Field.Index.ANALYZED));
				doc.add(new Field("text", text + " " + r[tw_text], Field.Store.YES, Field.Index.ANALYZED));
				doc.add(new Field("den", differenceDate(start,tweetTime)/numTweets+"", Field.Store.YES, Field.Index.NO));
				doc.add(new Field("n_retweets", Math.max(Integer.parseInt(r[tw_n_retweet]),n_retweets)+"", Field.Store.YES, Field.Index.NO));
				doc.add(new Field("pop", Math.max(pop,oldPop)+"", Field.Store.YES, Field.Index.NO));
				doc.add(new Field("start", start, Field.Store.YES, Field.Index.NO));
				doc.add(new Field("end", end, Field.Store.YES, Field.Index.NO));
				
				//System.out.print(writer.maxDoc());
				writer.deleteDocuments(idTerm);
				writer.addDocument(doc);
				//System.out.println(" --> " + writer.maxDoc());
			}
			return true;
		
	}
	
	/**
	 * @return true if user has been inserted 
	 */
	public boolean insertUser(String record) throws SQLException {
		String query;
		//System.out.println("user: " + record);
		String[] r = record.split("\t");
		// check if user is present
		query = "SELECT * FROM user WHERE id='"+r[usr_id]+"'";
		if (q) System.out.println("[Q] " + query);
		ResultSet rs = db.executeQuery(query);	
		if( rs!=null && rs.next() ) { 
			// update if necessary (if not update for one day or more)
			if (differenceDate(rs.getString("last_update"),now())>=3600*24) {
				query = "UPDATE user SET name = '"+formatString(r[usr_name])+"', screen_name = '"+formatString(r[usr_screen_name])+"', " +
				"image = '"+r[usr_image]+"', followers = '"+r[usr_followers]+"', last_update = '"+now()+"' "+
				"WHERE id = '"+r[usr_id]+"'";
				if (q) System.out.println("[Q] " + query);
				return db.executeUpdate(query);				
			}
		}
		else {
			// insert user
			query = "INSERT DELAYED INTO user (id, name, screen_name, image, followers, last_update) " +
			" VALUES ('"+r[usr_id]+"', '"+formatString(r[usr_name])+"', '"+formatString(r[usr_screen_name])+"', " +
			"'"+r[usr_image]+"', '"+r[usr_followers]+"', '"+now()+"')";
			if (q) System.out.println("[Q] " + query);
			return db.executeUpdate(query);
		}
		return false;
	}
	
	/** 
	 * Converts the GG MM HH:MM:SS YY CEST format into YYYY-MM-DD HH:MM:SS
	 */
	private String formatDate(String date) {
		String date_format;
		StringTokenizer st = new StringTokenizer(date, " ");
		String[] date_token = new String[st.countTokens()];
		
		int i = 0;
		while(st.hasMoreTokens())
		{
			String s = st.nextToken();
			date_token[i] = s;
			i++;
		}
		
		date_token[1] = date_token[1].replace("Jan", "01");
		date_token[1] = date_token[1].replace("Feb", "02");
		date_token[1] = date_token[1].replace("Mar", "03");
		date_token[1] = date_token[1].replace("Apr", "04");
		date_token[1] = date_token[1].replace("May", "05");
		date_token[1] = date_token[1].replace("Jun", "06");
		date_token[1] = date_token[1].replace("Jul", "07");
		date_token[1] = date_token[1].replace("Aug", "08");
		date_token[1] = date_token[1].replace("Sep", "09");
		date_token[1] = date_token[1].replace("Oct", "10");
		date_token[1] = date_token[1].replace("Nov", "11");
		date_token[1] = date_token[1].replace("Dec", "12");
		
		date_format = date_token[5]+"-"+date_token[1]+"-"+date_token[2]+" "+date_token[3];
		
		return date_format;
	}
	
	/**
	 * @return the difference in seconds between two dates
	 */
	private double differenceDate(String start, String finish) {
		double difference = 0;
		try {
			difference = (sdf.parse(finish).getTime() - sdf.parse(start).getTime())/1000;
		}
		catch (Exception e) {
			e.printStackTrace();
		}
		return difference;
	}
	
	/**
	 * @return the current datetime
	 */
	private String now() {
		String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
		Calendar cal = Calendar.getInstance();
		SimpleDateFormat sdf = new SimpleDateFormat(DATE_FORMAT);
		return sdf.format(cal.getTime());
	}
	
	 public static String formatString(String txt) {
		 return txt.replace("\\","\\\\").replace("\n"," ").replace("\t"," ").replace("'","\\'").replace("\"","\\\"").trim();
	 }
	
}